package competition_project

import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.{ALS, ALSModel}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, explode, to_json}
import org.apache.spark.sql.types.{ArrayType, FloatType, IntegerType, StructField, StructType}

import java.util.Properties

object recommended {
  def main(args: Array[String]): Unit = {
    /*
    1.显式反馈（Explicit Feedback）：例如用户给商品打分（1-5星），这种反馈是明确的、可量化的。
    2.隐式反馈（Implicit Feedback）：例如用户的点击、购买、浏览等行为，通常没有直接的评分。这类数据没有明确的“评分”信息，所以
    我们需要将其转化为某种可量化的形式。

    因为我们只有用户购买商品的数据，所以我们这里用到是隐式反馈
     */

    //  准备spark的环境
      val spark=SparkSession.builder()
        .master("local[*]")
        .appName("基于协同过滤的推荐")
        .getOrCreate()


    //  导入隐式转换，避免dataframe转化dataset的数据类型的不匹配的报错
    import spark.implicits._

    //  连接mysql的配置
    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  加载订单数据并选择需要用到的字段
    //  由于转换数据类型的时候会出现空值，下面做一个过滤
    val data=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/competition?useSSL=false","user_buy_info",connect)
      .select("user_id","product_id","buy_number")
      .withColumn("user_id",col("user_id").cast("long"))
      .withColumn("product_id",col("product_id").cast("long"))
      .withColumn("buy_number",col("buy_number").cast("long"))
      .filter(col("user_id").isNotNull)
      .filter(col("product_id").isNotNull)
      .filter(col("buy_number").isNotNull)

    //  检查是否存在空数据
//    data.filter(col("buy_number").isNull).show



    //  将数据分为两部分，一部分用来训练构建模型，一部分拿来测试
    val Array(training,test) =data.randomSplit(Array(0.8,0.2))

    //  构建als协同过滤的隐式反馈    implicit:隐式的
    /*
        setRegParam(0.1)：这个参数设置了 正则化参数（regParam），它用于控制模型复杂度，防止过拟合。在矩阵分解过程
        中，regParam 用于惩罚模型中的大权重值，从而避免模型过度依赖某些特定的特征，导致过拟合。setRegParam(0.1) 表示
        正则化项的系数为 0.1，较小的值允许模型有更多的自由度，较大的值则加强了正则化，限制了模型的复杂度。
     */
    val alsimplicit: ALS = new ALS()
      .setMaxIter(10) //  设置模型的最大迭代次数，次数越多模型的效果越好
      .setRegParam(0.1)
      .setImplicitPrefs(true)
      .setUserCol("user_id") //  用来作为用户的唯一标识符，als会为其构建用户特征向量
      .setItemCol("product_id") //  用来表示唯一的商品标识符,als会为其构建商品特征向量
      .setRatingCol("buy_number") //  将评分或者购买数量等构建用户和商品之间的关系
      .setColdStartStrategy("drop")   //  处理冷启动问题:删除没有足够数据的用户/商品,为了保证模型的准确性


    //  根据设置构建模型
    val model:ALSModel=alsimplicit.fit(training)

    //  predictions:预测
    val predictions=model.transform(test)

    //  查看预测的结果
    predictions.limit(5).show(false)


    //  对模型进行评估(使用均方根误差RMSE),构建评估器
    val evaluator=new RegressionEvaluator()
      .setMetricName("rmse")
      .setLabelCol("buy_number")
      .setPredictionCol("prediction")

    //  拿到评估的值，值越小，说明模型越好   evaluate:评估
    val rmse=evaluator.evaluate(predictions)
    print(s"评估的均方根值:${rmse}")

    //  给每个用户推荐前10个商品
    val user_recommend=model.recommendForAllUsers(10)
    //  查看每个用户推荐的10个商品
    user_recommend.limit(50).show(false)


    //  由于jdbc只支持基本的数据类型，不支持array<struct<product_id:int, rating:float>>:[{74, 0.59754586}, {215, 0.5829263}, {107, 0.529323},
    //  {108, 0.52268565}, {319, 0.5159624}, {81, 0.51302}, {232, 0.5121461}, {67, 0.5087154},
    //  {127, 0.5055256}, {257, 0.48255214}] 这样的复合数据类型，所以这一列的数据需要进行处理一下

    user_recommend.createOrReplaceTempView("result")

    val result_recommend=spark.sql(
      """
        |select
        |user_id,
        |replace(replace(cast(recommendations as string),"{",""),"}","") as recommendations
        |from result
        |""".stripMargin)

    //  将推荐的结果存放到mysql里面
    result_recommend.write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/competition?useSSL=false","result_recommend",connect)















    spark.close()
  }

}
